package com.youxin.kafka;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;

/**
 * Kafka消息制造
 * @author yingde.cao
 * 使用方法初始化扔进消息队列专用线程池来处理日志
 */
public class KafkaProducerTask implements Runnable{
	
	private static Logger logger=LoggerFactory.getLogger(KafkaProducerTask.class);
	private List<ProducerRecord<String, String>> messages;
    private Producer<String, String> producer;
    
	public KafkaProducerTask(Properties properties,List<ProducerRecord<String, String>> messages)
	{
		Objects.requireNonNull(properties, "properties cannot be null");
		Objects.requireNonNull(messages, "messages cannot be null");
		this.messages=messages;
        this.producer = new KafkaProducer<>(properties);
	}
	
	public KafkaProducerTask(Properties properties,ProducerRecord<String, String> message)
	{
		Objects.requireNonNull(properties, "properties cannot be null");
		Objects.requireNonNull(message, "message cannot be null");	
		this.messages=new ArrayList<>(1);
		this.messages.add(message);		
        this.producer = new KafkaProducer<>(properties);
	}
	
	@Override
	public void run() {
		try {
			for (ProducerRecord<String, String> message : messages) {				
				producer.send(message, (RecordMetadata recordMetadata, Exception e) -> {
					if (e != null) {						
						logger.error("生产消息异常", e);
					}
				});
			}			
		} catch (Exception e) {
			logger.error("生产消息异常", e);
		}
	}
	
//	public class Message{
//		private String key;
//		private String data;
//		public String getKey() {
//			return key;
//		}
//		public void setKey(String key) {
//			this.key = key;
//		}
//		public String getData() {
//			return data;
//		}
//		public void setData(String data) {
//			this.data = data;
//		}		
//	}
}
